/**
 * @author: dn-jinmin/dn-jinmin
 * @doc:
 */

package websocket

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/zeromicro/go-zero/core/threading"
	"time"

	"net/http"
	"sync"

	"github.com/gorilla/websocket"
	"github.com/zeromicro/go-zero/core/logx"
)

// AckType 是一个整型类型，用于定义消息所需确认级别。
// 它表示不同的确认级别以满足各种可靠性要求。
type AckType int

// Acknowledge 常量定义了三种消息确认级别。
const (
	// NoAck 表示消息不需要确认。
	// 适用于需要高吞吐量且可以牺牲可靠性的场景。
	NoAck AckType = iota
	// OnlyAck 表示需要确认，但不是严格保证。
	// 适用于需要中等吞吐量和可靠性的场景。
	OnlyAck
	// RigorAck 表示需要严格的确认，确保消息可靠接收。
	// 适用于消息可靠性至关重要的场景，即使会牺牲性能。
	RigorAck
)

// AckTypeToString 将确认类型转换为字符串表示。
// 该方法主要用于将 AckType 枚举类型的值转换为人类可读的字符串格式。
// 参数: 无
// 返回值: string 类型，表示 AckType 的字符串值。
func (t AckType) ToString() string {
	// 根据 AckType 的值选择相应的字符串返回。
	switch t {
	case OnlyAck:
		// 如果是 OnlyAck 类型，返回 "OnlyAck"。
		return "OnlyAck"
	case RigorAck:
		// 如果是 RigorAck 类型，返回 "RigorAck"。
		return "RigorAck"
	}

	// 如果 AckType 不是上述定义的任何一个值，返回 "NoAck"。
	return "NoAck"
}

// Server 结构体表示一个服务器实例，封装了服务器操作所需的功能和数据。
// 继承自 sync.RWMutex 以实现并发安全的读写操作。
type Server struct {
	sync.RWMutex

	opt            *serverOption  // 服务器选项配置
	authentication Authentication // 身份验证机制

	routes map[string]HandlerFunc // 路由表，映射路径到处理函数
	addr   string                 // 服务器监听地址
	patten string                 // URL 模式匹配规则

	connToUser map[*Conn]string // 连接到用户的映射
	userToConn map[string]*Conn // 用户到连接的映射

	upgrader    websocket.Upgrader // WebSocket 升级处理器
	logx.Logger                    // 日志记录器

	*threading.TaskRunner // 任务运行器
}

// NewServer 创建并返回一个新的 Server 实例。
// 参数：
//   - addr: 服务器监听地址
//   - opts: 可变数量的服务器选项配置
//
// 返回值：
//   - *Server: 新创建的服务器实例
func NewServer(addr string, opts ...ServerOptions) *Server {
	opt := newServerOptions(opts...) // 解析传入的服务器选项

	return &Server{
		routes:   make(map[string]HandlerFunc), // 初始化路由表
		addr:     addr,                         // 设置监听地址
		patten:   opt.patten,                   // 设置 URL 模式匹配规则
		opt:      &opt,                         // 设置服务器选项
		upgrader: websocket.Upgrader{},         // 初始化 WebSocket 升级处理器

		authentication: opt.Authentication, // 设置身份验证机制

		connToUser: make(map[*Conn]string), // 初始化连接到用户的映射
		userToConn: make(map[string]*Conn), // 初始化用户到连接的映射

		Logger:     logx.WithContext(context.Background()),   // 初始化日志记录器
		TaskRunner: threading.NewTaskRunner(opt.concurrency), // 初始化任务运行器
	}
}

// ServerWs 处理 WebSocket 连接请求
// 该方法升级HTTP连接为WebSocket连接，并进行身份验证
// 如果验证通过，将连接记录并交由handlerConn方法处理
func (s *Server) ServerWs(w http.ResponseWriter, r *http.Request) {
	// 捕获并处理方法执行过程中的panic
	defer func() {
		if r := recover(); r != nil {
			s.Errorf("server handler ws recover err %v", r)
		}
	}()

	// 创建新的WebSocket连接
	conn := NewConn(s, w, r)
	// 如果连接创建失败，直接返回
	if conn == nil {
		return
	}

	// 进行身份验证
	if !s.authentication.Auth(w, r) {
		// 如果验证失败，发送错误消息并关闭连接
		s.Send(&Message{FrameType: FrameData, Data: fmt.Sprint("不具备访问权限")}, conn)
		conn.Close()
		return
	}

	// 记录连接
	s.addConn(conn, r)

	// 处理连接
	go s.handlerConn(conn)
}

// 根据连接对象执行任务处理
func (s *Server) handlerConn(conn *Conn) {
	// 获取用户ID列表
	uids := s.GetUsers(conn)
	// 为连接分配第一个用户ID
	conn.Uid = uids[0]

	// 启动一个协程处理写入操作
	go s.handlerWrite(conn)

	// 如果是ACK类型的消息，启动一个协程读取ACK
	if s.isAck(nil) {
		go s.readAck(conn)
	}

	for {
		// 获取请求消息
		_, msg, err := conn.ReadMessage()
		if err != nil {
			// 如果读取消息出错，记录错误并关闭连接
			s.Errorf("websocket conn read message err %v", err)
			s.Close(conn)
			return
		}
		// 解析消息
		var message Message
		if err = json.Unmarshal(msg, &message); err != nil {
			// 如果消息解析出错，记录错误并关闭连接
			s.Errorf("json unmarshal err %v, msg %v", err, string(msg))
			s.Close(conn)
			return
		}

		// 依据消息进行处理
		if s.isAck(&message) {
			// 如果是ACK消息，记录信息并添加到消息队列
			s.Infof("conn message read ack msg %v", message)
			conn.appendMsgMq(&message)
		} else {
			// 否则，将消息发送到处理通道
			conn.message <- &message
		}
	}
}

// isAck 判断消息是否需要确认
// 如果消息为nil，则根据服务器配置决定是否需要确认
// 如果消息不为nil，则根据消息的帧类型和服务器配置决定是否需要确认
func (s *Server) isAck(message *Message) bool {
	if message == nil {
		return s.opt.ack != NoAck
	}
	return s.opt.ack != NoAck && message.FrameType != FrameNoAck
}

// readAck 读取消息的确认
// 该方法在一个循环中不断检查连接是否已关闭以及是否有消息需要确认
// 如果连接已关闭，则退出循环；否则，根据服务器配置的确认模式处理消息
func (s *Server) readAck(conn *Conn) {
	for {
		select {
		case <-conn.done:
			s.Infof("close message ack uid %v ", conn.Uid)
			return
		default:
		}

		// 从队列中读取新的消息
		conn.messageMu.Lock()
		if len(conn.readMessage) == 0 {
			conn.messageMu.Unlock()
			// 增加睡眠
			time.Sleep(100 * time.Microsecond)
			continue
		}

		// 读取第一条
		message := conn.readMessage[0]

		// 判断ack的方式
		switch s.opt.ack {
		case OnlyAck:
			// 直接给客户端回复
			s.Send(&Message{
				FrameType: FrameAck,
				Id:        message.Id,
				AckSeq:    message.AckSeq + 1,
			}, conn)
			// 进行业务处理
			// 把消息从队列中移除
			conn.readMessage = conn.readMessage[1:]
			conn.messageMu.Unlock()

			conn.message <- message
		case RigorAck:
			// 先回
			if message.AckSeq == 0 {
				// 还未确认
				conn.readMessage[0].AckSeq++
				conn.readMessage[0].ackTime = time.Now()
				s.Send(&Message{
					FrameType: FrameAck,
					Id:        message.Id,
					AckSeq:    message.AckSeq,
				}, conn)
				s.Infof("message ack RigorAck send mid %v, seq %v , time%v", message.Id, message.AckSeq,
					message.ackTime)
				conn.messageMu.Unlock()
				continue
			}

			// 再验证

			// 1. 客户端返回结果，再一次确认
			// 得到客户端的序号
			msgSeq := conn.readMessageSeq[message.Id]
			if msgSeq.AckSeq > message.AckSeq {
				// 确认
				conn.readMessage = conn.readMessage[1:]
				conn.messageMu.Unlock()
				conn.message <- message
				s.Infof("message ack RigorAck success mid %v", message.Id)
				continue
			}

			// 2. 客户端没有确认，考虑是否超过了ack的确认时间
			val := s.opt.ackTimeout - time.Since(message.ackTime)
			if !message.ackTime.IsZero() && val <= 0 {
				//		2.2 超过结束确认
				delete(conn.readMessageSeq, message.Id)
				conn.readMessage = conn.readMessage[1:]
				conn.messageMu.Unlock()
				continue
			}
			//		2.1 未超过，重新发送
			conn.messageMu.Unlock()
			s.Send(&Message{
				FrameType: FrameAck,
				Id:        message.Id,
				AckSeq:    message.AckSeq,
			}, conn)
			// 睡眠一定的时间
			time.Sleep(3 * time.Second)
		}
	}
}

// 任务的处理
// 该函数负责处理写入连接的消息，根据消息类型进行不同的处理
func (s *Server) handlerWrite(conn *Conn) {
	for {
		select {
		case <-conn.done:
			// 连接关闭
			return
		case message := <-conn.message:
			switch message.FrameType {
			case FramePing:
				// 处理Ping帧
				s.Send(&Message{FrameType: FramePing}, conn)
			case FrameData:
				// 根据请求的method分发路由并执行
				if handler, ok := s.routes[message.Method]; ok {
					handler(s, conn, message)
				} else {
					// 如果不存在对应的方法，发送错误信息
					s.Send(&Message{FrameType: FrameData, Data: fmt.Sprintf("不存在执行的方法 %v 请检查", message.Method)}, conn)
				}
			}
			// 如果消息需要确认，将其从待确认队列中删除
			if s.isAck(message) {
				conn.messageMu.Lock()
				delete(conn.readMessageSeq, message.Id)
				conn.messageMu.Unlock()
			}
		}
	}
}

// 添加连接
// 该函数负责将新连接添加到服务器，并处理用户与连接的映射
func (s *Server) addConn(conn *Conn, req *http.Request) {
	uid := s.authentication.UserId(req)

	s.RWMutex.Lock()
	defer s.RWMutex.Unlock()

	// 验证用户是否之前登入过
	if c := s.userToConn[uid]; c != nil {
		// 关闭之前的连接
		c.Close()
	}

	// 建立用户与新连接的映射
	s.connToUser[conn] = uid
	s.userToConn[uid] = conn
}

// 获取连接
// 根据用户ID获取对应的连接
func (s *Server) GetConn(uid string) *Conn {
	s.RWMutex.RLock()
	defer s.RWMutex.RUnlock()

	return s.userToConn[uid]
}

// 获取多个连接
// 根据多个用户ID获取对应的连接
func (s *Server) GetConns(uids ...string) []*Conn {
	if len(uids) == 0 {
		return nil
	}

	s.RWMutex.RLock()
	defer s.RWMutex.RUnlock()

	res := make([]*Conn, 0, len(uids))
	for _, uid := range uids {
		res = append(res, s.userToConn[uid])
	}
	return res
}

// 获取用户
// 根据连接获取对应的用户ID，或获取所有用户ID
func (s *Server) GetUsers(conns ...*Conn) []string {
	s.RWMutex.RLock()
	defer s.RWMutex.RUnlock()

	var res []string
	if len(conns) == 0 {
		// 获取全部
		res = make([]string, 0, len(s.connToUser))
		for _, uid := range s.connToUser {
			res = append(res, uid)
		}
	} else {
		// 获取部分
		res = make([]string, 0, len(conns))
		for _, conn := range conns {
			res = append(res, s.connToUser[conn])
		}
	}

	return res
}

// 关闭连接
// 根据连接对象关闭连接，并更新用户与连接的映射
func (s *Server) Close(conn *Conn) {
	s.RWMutex.Lock()
	defer s.RWMutex.Unlock()

	uid := s.connToUser[conn]
	if uid == "" {
		// 已经被关闭
		return
	}

	// 更新映射并关闭连接
	delete(s.connToUser, conn)
	delete(s.userToConn, uid)

	conn.Close()
}

// 通过用户ID发送消息
// 根据用户ID获取连接，并向这些连接发送消息
func (s *Server) SendByUserId(msg interface{}, sendIds ...string) error {
	if len(sendIds) == 0 {
		return nil
	}

	return s.Send(msg, s.GetConns(sendIds...)...)
}

// 发送消息
// 向指定的连接发送消息
func (s *Server) Send(msg interface{}, conns ...*Conn) error {
	if len(conns) == 0 {
		return nil
	}

	data, err := json.Marshal(msg)
	if err != nil {
		return err
	}

	for _, conn := range conns {
		if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
			return err
		}
	}

	return nil
}

// 添加路由
// 为服务器添加新的路由规则
func (s *Server) AddRoutes(rs []Route) {
	for _, r := range rs {
		s.routes[r.Method] = r.Handler
	}
}

// 启动服务器
// 设置HTTP处理函数并启动服务器
func (s *Server) Start() {
	http.HandleFunc(s.patten, s.ServerWs)
	s.Info(http.ListenAndServe(s.addr, nil))
}

// 停止服务器
// 打印停止服务的消息
func (s *Server) Stop() {
	fmt.Println("停止服务")
}
